package cn.doitedu.rtdw.exercise;

import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Ex01_bitmap轻度聚合练习 {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(1000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");
        env.getCheckpointConfig().setCheckpointTimeout(1000);
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);


        // 源表逻辑映射表
        tenv.executeSql("create table order_kafka(\n" +
                "    user_id bigint,\n" +
                "    product_cat string,\n" +
                "    product_brand string,\n" +
                "    order_id string,\n" +
                "    order_amt decimal(10,2),\n" +
                "    create_time bigint,\n" +
                "    rt as to_timestamp_ltz(create_time,3),\n" +
                "    watermark for rt as rt - interval '0' second \n" +
                ") with (\n" +
                "    'connector' = 'kafka',\n" +
                "    'topic' = 'topic_order',\n" +
                "    'properties.bootstrap.servers' = 'doitedu:9092',\n" +
                "    'properties.group.id' = 'goo3',\n" +
                "    'scan.startup.mode' = 'latest-offset',\n" +
                "    'value.format'='json',\n" +
                "    'value.json.fail-on-missing-field'='false',\n" +
                "    'value.fields-include' = 'EXCEPT_KEY'\n" +
                ")");

        // doris目标表的逻辑映射表
        tenv.executeSql("create table order_doris(\n" +
                "    product_cat    varchar(20),\n" +
                "    product_brand  varchar(20),\n" +
                "    order_count    bigint,\n" +
                "    order_amt      decimal(10,2),\n" +
                "    customer_cnt   bigint \n" +
                ") WITH (                                                                                        \n" +
                "   'connector' = 'doris',                                                                       \n" +
                "   'fenodes' = 'doitedu:8030',                                                                  \n" +
                "   'table.identifier' = 'test.order_agg',                                               \n" +
                "   'username' = 'root',                                                                         \n" +
                "   'password' = 'root',                                                                         \n" +
                "   'sink.label-prefix' = 'doit42-02-3',                                                         \n" +
                "   'sink.properties.columns' = 'product_cat,product_brand,order_count,order_amt,customer_cnt,customer_cnt=to_bitmap(customer_cnt)'                                    \n" +
                ")");


        // 统计sql
        /**
         *
         * +----+--------------+----------------+----------------+--------------+--------------+
         * | op |  product_cat |  product_brand |    order_count |    order_amt | customer_cnt |
         * +----+--------------+----------------+----------------+--------------+--------------+
         * | +I |        cat01 |        brand01 |              2 |       300.00 |            1 |
         * | +I |        cat01 |        brand01 |              2 |       400.00 |            3 |
         * | +I |        cat01 |        brand01 |              1 |       250.00 |            2 |
         *
         *
         */
        tenv.executeSql("insert into order_doris  " +
                "select                           \n" +
                "    product_cat,                 \n" +
                "    product_brand,               \n" +
                "    count(1) as order_count,     \n" +
                "    sum(order_amt) as order_amt, \n" +
                "    user_id as customer_cnt      \n" +
                "from table(           \n" +
                "    tumble(table order_kafka,descriptor(rt),interval '1' minute)\n" +
                ")                     \n" +
                "group by              \n" +
                "    window_start,     \n" +
                "    window_end,       \n" +
                "    user_id,          \n" +
                "    product_cat,      \n" +
                "    product_brand");





    }

}
